package com.gvtv.main.execute;

import java.util.Date;
import java.util.List;

import org.apache.commons.lang.StringUtils;
import org.apache.log4j.Logger;
import org.apache.thrift.TException;
import org.joda.time.DateTime;

import com.gvtv.main.exception.ExecuteException;
import com.gvtv.main.execute.ExecuteJob.ExecuteJobState;
import com.gvtv.main.hive.HiveExecute;
import com.gvtv.main.hive.HiveJdbc;
import com.gvtv.main.schedul.SchedulEnv;
import com.gvtv.main.schedul.SchedulInfo;
import com.gvtv.main.util.AppTools;
import com.gvtv.main.util.Constant;

public class HiveJobWorker extends AbstractJobWorker implements JobWorker{

	private Logger logger = Logger.getLogger(HiveJobWorker.class);
	
	public HiveJobWorker(SchedulInfo schedulInfo, ExecuteJob executeJob, SchedulEnv env) {
		super(schedulInfo, executeJob, env);
	}

	@Override
	public void work() {
		job.setStartTime(new Date());
		logger.info("job ["+job.getId()+"]开始运行，运行时间："+new DateTime(job.getSubmitTime().getTime()).toString(Constant.DATETIME_FORMAT));
		env.convert(job);
		if(StringUtils.isBlank(job.getScript())){
			job.setJobState(ExecuteJobState.FAILED);
			job.setEndTime(new Date());
			logger.error("job ["+job.getId()+"]sql语句为空，运行失败，结束时间："+job.getEndTime());
			return;
		}
		List<String> args = job.getArgs();
		if(args!=null && args.size()>0){//判断是否存在<param name="" type="arg"></param>配置，存在则需要通过hivejdbc 参数的方式运行sql语句
			HiveJdbc hiveJdbc = new HiveJdbc();
			List<String> sets = job.getHiveSets();
			String sql = AppTools.replaceBlank(job.getScript());
			try {
				if(sets!=null && !sets.isEmpty()){
					for (String set : sets) {
						if(StringUtils.isBlank(set)){
							logger.warn("job ["+job.getId()+"] 的hive set 语句为空，跳过。。。");
							continue;
						}
						hiveJdbc.execute(AppTools.replaceBlank(set));
					}
				}
				
				logger.info("job ["+job.getId()+"] 执行sql: " + sql);
				hiveJdbc.execute(sql, args.toArray(new String[]{}));
				job.setJobState(ExecuteJobState.SUCCESS);
			} catch (ExecuteException e) {
				logger.error("job ["+job.getId()+"] 执行sql["+sql+"]失败。" + e);
				job.setJobState(ExecuteJobState.FAILED);
			}
			hiveJdbc.close();
		}else {//不存在<param name="" type="arg"></param>配置，则选择hiveexecute方式运行sql语句
			String[] sqls = AppTools.replaceBlank(job.getScript()).split(";");
			HiveExecute hiveExecute = new HiveExecute();
			for (String sql : sqls) {
				if(StringUtils.isBlank(sql)){
					logger.warn("job ["+job.getId()+"] sql语句为空，跳过。。。");
					continue;
				}
				try {
					logger.info("job ["+job.getId()+"] 执行sql: " + sql);
					hiveExecute.execute(sql);
					job.setJobState(ExecuteJobState.SUCCESS);
				} catch (TException e) {
					logger.error("job ["+job.getId()+"] 执行sql["+sql+"]失败。" + e);
					job.setJobState(ExecuteJobState.FAILED);
					break;
				}
			}
			hiveExecute.close();
		}
		job.setEndTime(new Date());
		logger.info("job ["+job.getId()+"]结束运行，结束时间："+new DateTime(job.getEndTime().getTime()).toString(Constant.DATETIME_FORMAT)+"，运行状态，"+job.getJobState());
	}
}
